package com.at.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 *
 * @author cdhuangchao3
 * @date 2023/5/29 9:24 PM
 */
object SparkStreaming07_Output1 {

  def main(args: Array[String]): Unit = {
    // 创建环境
    // 创建时，需要传递2个参数：
    //    param： 环境配置
    val sc = new SparkConf().setMaster("local[*]").setAppName("operator")
    //    param2: 采集周期
    val ssc = new StreamingContext(sc, Seconds(3))
    ssc.checkpoint("cp")

    val lines = ssc.socketTextStream("localhost", 9999)

    val word2One = lines.map((_, 1))

    // reduceByKeyAndWindow: 当窗口范围比较大，但是滑动幅度比较小，那么可以采用增加数据和删除数据的方式
    //    无需重新计算，提升性能
    val windowDS = word2One.reduceByKeyAndWindow(
      (x: Int, y: Int) => {
        x + y
      },
      (x: Int, y: Int) => {
        x - y
      },
      Seconds(9),
      Seconds(3)
    )
    // foreachRDD 不会出现时间戳
    windowDS.foreachRDD(
      rdd => {

      }
    )

    // 1、启动采集器
    ssc.start()
    // 2、等待采集器的关闭
    ssc.awaitTermination()
  }

}
